package org.softee.concurrent;

import static org.softee.concurrent.SequentialConcurrentService.State.CANCELED;
import static org.softee.concurrent.SequentialConcurrentService.State.FAILED;
import static org.softee.concurrent.SequentialConcurrentService.State.NEW;
import static org.softee.concurrent.SequentialConcurrentService.State.RUNNING;
import static org.softee.concurrent.SequentialConcurrentService.State.STARTING;
import static org.softee.concurrent.SequentialConcurrentService.State.STOPPED;
import static org.softee.concurrent.SequentialConcurrentService.State.TERMINATED;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

import javax.management.MalformedObjectNameException;

import org.softee.management.exception.ManagementException;

/**
 * This service is useful for processing that must produce results in the same order in which the tasks are submitted,
 * but still being able to run exploiting all available CPU resources to perform the processing.
 * It is convenient for many batch processing type tasks.<p>
 *
 * Processing design:<p>
 * <ol>
 * <li>Tasks submitted to submit(Callable) are queued on a taskQueue</li>
 * <li>A (single threaded) task service takes takes callables from the task queue, submits them to a (concurrent)
 * service</li>
 * <li>The resulting Future holding the result is immediately queued on a result queue</li>
 * <li>A (single threaded) result service takes the Futures from the result queue, gets the value, waiting if necessary,
 * and asks the result listener to accept() the result.
 *
 * @author morten.hattesen@gmail.com
 *
 * @param <T> type of the results of the callables submitted to this service
 */

public class SequentialConcurrentService <T> {
    public enum State {NEW, STARTING, RUNNING, STOPPED, CANCELED, FAILED, TERMINATED};
    final AtomicReference<State> state = new AtomicReference<State>(State.NEW);
    final int availableProcessors;
    final int threads;
    final int taskQueueSize;
    final ExecutorService service;
    final BlockingQueue<Future<T>> taskQueue;
    final ExecutorService resultService = Executors.newSingleThreadExecutor();
    final ResultListener<T> resultListener;
    volatile Future<Void> resultServiceFuture;
    private Runnable doNothing = new Runnable() {
        public void run() {}
    };
    private Future<T> endFuture = new FutureTask<T>(doNothing, null);

    public SequentialConcurrentService(ResultListener<T> resultListener) throws MalformedObjectNameException {
        this(resultListener, 1.75, -10);
    }

    /**
     * @param resultListener A listener that will be notified of completed results, and notified when processing has completed
     * @param concurrencyFactor The number of threads that may concurrently process tasks for every CPU core available on the system.<br>
     * Choose lower values, if tasks are CPU bound.<br>
     * Choose higher values if tasks are resource (I/O) bound.<br>
     * A maximum of ({@code concurrencyFactor} * {@code availableCpuCores}) threads will be created.
     * Typical values: 1 .. 10.<br>
     * @param taskQueueSizeHint the size of the queue holding tasks being concurrently processed.<br>
     * negative value (-n) to indicate that the queue should be sized as {@code n} * number of threads.<br>
     * The value 0 specifies unbounded queues.<br>
     * Choose lower values if tasks are memory intensive.<br>
     * Choose higher values if tasks are not homogenous, i.e. vary in how much processing resources they require to complete.
     * Typical values: 2 .. 100 (or rather -2 .. -100)
     * @throws MalformedObjectNameException
     */
    public SequentialConcurrentService(ResultListener<T> resultListener, double concurrencyFactor, int taskQueueSizeHint) throws MalformedObjectNameException {
        this.resultListener = resultListener;
        availableProcessors = Runtime.getRuntime().availableProcessors();
        threads = (int) (availableProcessors * concurrencyFactor);
        this.taskQueueSize = taskQueueSizeHint >= 0 ? taskQueueSizeHint : -taskQueueSizeHint * threads;
        service = Executors.newFixedThreadPool(threads);
        taskQueue = (taskQueueSize > 0) ?
                new ArrayBlockingQueue<Future<T>>(taskQueueSize)
                : new LinkedBlockingQueue<Future<T>>();
    }

    /**
     * Start the concurrent service.<p>
     *
     * The method will return immediately
     * @throws ManagementException
     */
    public void start() throws ManagementException  {
        if (!state.compareAndSet(NEW, STARTING)) {
            throw new IllegalStateException("Unable to start service with state " + state);
        }

        resultServiceFuture = resultService.submit(new Callable<Void>() {
            public Void call() throws Exception {
                try {
                    while (state.get() != CANCELED) {
                        Future<T> resultFuture = taskQueue.take();
                        if (resultFuture == endFuture) {
                            break; // End marker detected
                        }
                        T result = resultFuture.get(); // this may throw an ExecutionException
                        resultListener.accept(result);
                    }
                    // all results were processed, tell the listener to close
                    resultListener.close();
                    // bail out
                } catch (Exception e) {
                    cancel(); // no point in producing more tasks
                    resultListener.notifyFailure(e);
                    state.compareAndSet(CANCELED, FAILED);
                    throw e; // rethrow (caught by getting the value of the future
                }
                return null;
            }
        });

        if (!state.compareAndSet(STARTING, RUNNING)) {
            // may have been canceled by another thread
            throw new IllegalStateException("Unable to start service, as it is " + state);
        }
    }

    public void submit(Callable<T> task) throws InterruptedException {
        if (state.get() != RUNNING) {
            throw new IllegalStateException("Unable to submit task on service with state " + state);
        }
        Future<T> result = service.submit(task);
        // This may block
        taskQueue.put(result);
    }

    /**
     * Wait for completion of all pending processing and return.
     * The method will return when all processing has been completed, if an exception occurred
     * or if the service was canceled (by another thread).<p>
     * @throws InterruptedException if the thread was interrupted while waiting for the services to complete
     * @throws ExecutionException if either the submitted callables or the listener
     * @throws ManagementException
     */
    public void stopAndWait() throws InterruptedException, ExecutionException, ManagementException {
        stop();
        waitForCompletion();
    }

    public void stop() throws InterruptedException {
        // indicate that no further input is expected
        if (!state.compareAndSet(RUNNING, STOPPED)) {
            throw new IllegalStateException("Unable to stop service with state " + state);
        }
        taskQueue.put(endFuture);
    }

    public void waitForCompletion() throws InterruptedException, ExecutionException {
        if (state.get() != STOPPED) {
            throw new IllegalStateException("Unable to wait for completion of service with state " + state);
        }
        // wait for the resultService to complete, throwing an ExecutionException if failed
        resultServiceFuture.get();

        service.shutdown();
        resultService.shutdown();
        state.set(TERMINATED);
    }

    protected void cancel() {
        if (state.get() == CANCELED || state.get() == TERMINATED) {
            throw new IllegalStateException("Unable to cancel service with state " + state);
        }
        state.set(CANCELED);
        service.shutdownNow(); // no point in producing more results
        resultService.shutdownNow(); // no point in delivering more results
    }
}
